﻿using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Hosting;

namespace NetMicro.Cache.DelayBus
{
    public class RedisDelayHostService : BackgroundService
    {
        /// <summary>
        /// 延迟队列频道信息
        /// </summary>
        private static readonly ConcurrentBag<string> DelayQueues = new ConcurrentBag<string>();
        /// <summary>
        /// 延迟消息处理接口
        /// </summary>
        private readonly IDelayedMessageProcessor _delayedMessageProcessor;

        public RedisDelayHostService(IDelayedMessageProcessor delayedMessageProcessor)
        {
            _delayedMessageProcessor = delayedMessageProcessor;
        }
        protected override Task ExecuteAsync(CancellationToken stoppingToken)
        {
            ConcurrentBag<string> DelayTopics = DelayTopic.Get();
            foreach (string topic in DelayTopics)
            {
                if (string.IsNullOrWhiteSpace(topic))
                {
                    throw new ArgumentException($"异常的Topic");
                }
                //避免重复注册
                if (!DelayQueues.Contains(topic))
                {
                    DelayQueues.Add(topic);
                    Task.Factory.StartNew(() =>
                     {
                         _delayedMessageProcessor.DeliveryToReadyQueueAsync(topic);
                     }, TaskCreationOptions.LongRunning);

                    Task.Factory.StartNew(() =>
                    {
                        _delayedMessageProcessor.ConsumeReadyJobAsync(topic);
                    }, TaskCreationOptions.LongRunning);

                }

            }

            return Task.CompletedTask;
        }
    }
}
